package org.apache.rocketmq.namesrv.routeinfo;

import io.netty.channel.Channel;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.common.MixAll;
import org.apache.rocketmq.common.TopicConfig;
import org.apache.rocketmq.common.constant.LoggerName;
import org.apache.rocketmq.common.constant.PermName;
import org.apache.rocketmq.common.namesrv.NamesrvConfig;
import org.apache.rocketmq.common.topic.TopicValidator;
import org.apache.rocketmq.common.utils.ConcurrentHashMapUtils;
import org.apache.rocketmq.logging.org.slf4j.Logger;
import org.apache.rocketmq.logging.org.slf4j.LoggerFactory;
import org.apache.rocketmq.namesrv.NamesrvController;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.apache.rocketmq.remoting.exception.RemotingConnectException;
import org.apache.rocketmq.remoting.exception.RemotingSendRequestException;
import org.apache.rocketmq.remoting.exception.RemotingTimeoutException;
import org.apache.rocketmq.remoting.exception.RemotingTooMuchRequestException;
import org.apache.rocketmq.remoting.protocol.DataVersion;
import org.apache.rocketmq.remoting.protocol.RemotingCommand;
import org.apache.rocketmq.remoting.protocol.RequestCode;
import org.apache.rocketmq.remoting.protocol.body.ClusterInfo;
import org.apache.rocketmq.remoting.protocol.body.TopicConfigSerializeWrapper;
import org.apache.rocketmq.remoting.protocol.header.NotifyMinBrokerIdChangeRequestHeader;
import org.apache.rocketmq.remoting.protocol.header.namesrv.UnRegisterBrokerRequestHeader;
import org.apache.rocketmq.remoting.protocol.namesrv.RegisterBrokerResult;
import org.apache.rocketmq.remoting.protocol.route.BrokerData;
import org.apache.rocketmq.remoting.protocol.route.QueueData;
import org.apache.rocketmq.remoting.protocol.route.TopicRouteData;

import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;


/**
 * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
 * @author：陈清风扬，个人微信号：chenqingfengyangjj。
 * @date:2025/1/10
 * @方法描述：路由信息管理器，这是Namesrv模块最核心的一个组件，用于管理主题路由信息，包括broker地址、集群地址、broker活跃信息、过滤服务器信息等。
 */
public class RouteInfoManager {


    private static final Logger log = LoggerFactory.getLogger(LoggerName.NAMESRV_LOGGER_NAME);

    //默认的Broker和Namesrv构建的channel的过期时间
    private final static long DEFAULT_BROKER_CHANNEL_EXPIRED_TIME = 1000 * 60 * 2;

    private final ReadWriteLock lock = new ReentrantReadWriteLock();

    //这个成员存储的是BrokerName和BrokerData的映射关系，其实就是存储了Broker集群和集群中所有Broker信息的键值对
    //因为同一个Broker集群中的Broker节点的名称是相同的，所以这里使用BrokerName作为key
    private final Map<String/* brokerName */, BrokerData> brokerAddrTable;

    //这个成员变量存储了集群名称和集群中所有Broker节点名称的映射关系
    //这里大家可能会困惑，不是说一个集群下的Broker节点的名称是相同的吗？为什么这里还要存储集群名称和集群中所有Broker节点名称的映射关系呢？
    //如果又要给Test集群，集群下有两个Broker节点，一个是主节点一个是从节点，名称都是Broker-a，主节点Id为0，从节点为1。这种情况下这里的value中存储的名称不就只有一个Broker-a吗？
    //确实是这样，如果一个集群中只有一个主节点，那么这个集群中的所有Broker节点的名称都是相同的，但是如果一个集群中有多个主节点，那么这个集群中的所有Broker节点的名称就不一样了
    //就比如说构建一个双柱双从的Broker集群。Broker主节点的名称分别为Broker-a、Broker-b，都在Test集群下，这时候这个value中存储的就是Broker-a和Broker-b了
    //但我们目前还不必考虑那么远的情况，现在我们就只考虑一个集群中只有一个主节点的情况即可
    private final Map<String/* clusterName */, Set<String/* brokerName */>> clusterAddrTable;

    //这个成员变量存储的是每一个注册到Namesrv节点的Broker节点的活跃信息，BrokerLiveInfo包括最后一次更新时间、心跳超时时间、数据版本、通道信息、高可用服务器地址等
    //BrokerAddrInfo中存储了Broker节点的名称和所在集群的名称
    private final Map<BrokerAddrInfo/* brokerAddr */, BrokerLiveInfo> brokerLiveTable;


    //这个成员变量存储的是主题和队列信息的映射关系，这个映射关系是一个多级映射关系，第一级key是主题名称，第二级key是Broker节点的名称，value是QueueData对象
    private final Map<String/* topic */, Map<String, QueueData>> topicQueueTable;
    //批量注销Broker信息的成员变量
    private final BatchUnregistrationService unRegisterService;

    private final NamesrvController namesrvController;
    private final NamesrvConfig namesrvConfig;


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @date:2025/1/10
     * @方法描述：构造方法
     */
    public RouteInfoManager(final NamesrvConfig namesrvConfig, NamesrvController namesrvController) {
        this.topicQueueTable = new ConcurrentHashMap<>(1024);
        this.brokerAddrTable = new ConcurrentHashMap<>(128);
        this.clusterAddrTable = new ConcurrentHashMap<>(32);
        this.brokerLiveTable = new ConcurrentHashMap<>(256);
        this.unRegisterService = new BatchUnregistrationService(this, namesrvConfig);
        this.namesrvConfig = namesrvConfig;
        this.namesrvController = namesrvController;
    }


    //启动路由信息管理器的方法
    public void start() {
        //启动批量注销Broker节点信息的服务
        this.unRegisterService.start();
    }

    //关闭路由信息管理器的方法
    public void shutdown() {
        //关闭批量注销Broker节点信息的服务
        this.unRegisterService.shutdown(true);
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @date:2025/1/10
     * @方法描述：注册Broker节点信息到路由信息管理器中的方法
     */
    public RegisterBrokerResult registerBroker(
            final String clusterName,
            final String brokerAddr,
            final String brokerName,
            final long brokerId,
            final String haServerAddr,
            final String zoneName,
            final Long timeoutMillis,
            final Boolean enableActingMaster,
            final TopicConfigSerializeWrapper topicConfigWrapper,
            final List<String> filterServerList,
            final Channel channel) {
        //创建注册结果对象，用来返回给Broker节点
        RegisterBrokerResult result = new RegisterBrokerResult();
        try {
            this.lock.writeLock().lockInterruptibly();
            //判断clusterAddrTable成员变量中是否存储着当前注册过来的Broker节点所在集群的信息，如果没有存储，那就创建clusterName-> new HashSet<>()键值对存储到clusterAddrTable中
            Set<String> brokerNames = ConcurrentHashMapUtils.computeIfAbsent((ConcurrentHashMap<String, Set<String>>) this.clusterAddrTable, clusterName, k -> new HashSet<>());
            //把当前注册过来的Broker节点的名称存储到brokerNames集合中
            brokerNames.add(brokerName);
            //定义一个是否第一次注册的标志
            boolean registerFirst = false;
            //根据当前注册过来的Broker节点的名称获取BrokerData对象
            //这里得到的其实就是Broker所在集群的全部节点信息
            BrokerData brokerData = this.brokerAddrTable.get(brokerName);
            //如果brokerData为空，说明当前注册的Broker节点是第一次注册
            //确切地说是Broker集群中的第一个注册过来的Broker节点
            if (null == brokerData) {
                //这时候就把对应的BrokerData对象创建出来
                brokerData = new BrokerData(clusterName, brokerName, new HashMap<>());
                //把BrokerData对象存储到brokerAddrTable中
                this.brokerAddrTable.put(brokerName, brokerData);
            }
            //如果当前注册过来的Broker节点没有启用ActingMaster模式，意味着当前注册的Broker节点是旧版本的Broker节点
            boolean isOldVersionBroker = enableActingMaster == null;
            //设置Broker是否启用了ActingMaster模式，在第五版本代码中，大家肯定都不清楚这个ActingMaster模式是什么
            //这个是MQ新版本添加的一个功能，在Broker集群状态下会用到，如果有朋友想提前了解一下，可以看看RocketMq的官方文档
            //或者直接去github上查看源码doc模块下的文档，里面有一篇Slave Acting Master文档专门讲解了这个ActingMaster模式的作用
            brokerData.setEnableActingMaster(!isOldVersionBroker && enableActingMaster);
            //设置Broker节点所在区域的名称
            brokerData.setZoneName(zoneName);
            //从brokerData对象中获取Broker节点的地址信息
            //这里得到的是一个map，map的key就是Broker节点的Id，value就是Broker节点的地址
            //我之前也说过，BrokerData存储的是一个Broker集群中所有节点的信息，这个信息包括节点的名称、节点的Id、节点的地址等
            //而Broker集群中每一个节点的Id就存放在这个brokerAddrsMap中，是map的key，value就是节点的地址
            Map<Long, String> brokerAddrsMap = brokerData.getBrokerAddrs();
            //以下这行代码可以转换成这样写，更加直观
            //Iterator<Map.Entry<Long, String>> iterator = brokerAddrsMap.entrySet().iterator();
            //while (iterator.hasNext()) {
            //    Map.Entry<Long, String> item = iterator.next();
            //    if (brokerAddr != null && brokerAddr.equals(item.getValue()) && brokerId != item.getKey()) {
            //        iterator.remove();
            //    }
            //}

            boolean isMinBrokerIdChanged = false;
            long prevMinBrokerId = 0;
            //如果brokerAddrsMap不为空，获取brokerAddrsMap中的最小的Broker节点的Id
            if (!brokerAddrsMap.isEmpty()) {
                prevMinBrokerId = Collections.min(brokerAddrsMap.keySet());
            }
            //如果新注册的Broker节点的Id小于brokerAddrsMap中存储的最小的Broker节点的Id，那就说明最小的Broker节点的Id发生了变化
            //Id越小，意味着节点越有可能是Broker集群的主节点
            if (brokerId < prevMinBrokerId) {
                isMinBrokerIdChanged = true;
            }
            //先判断brokerAddrsMap中是否存储着当前注册过来的Broker节点的地址信息，如果存储了，并且当前注册过来的Broker节点的Id和brokerAddrsMap中存储的Id不一致
            //那就把brokerAddrsMap中存储的这个Broker节点的信息删除掉，把最新的注册过来的Broker节点的Id和地址信息存储进去
            brokerAddrsMap.entrySet().removeIf(item -> null != brokerAddr && brokerAddr.equals(item.getValue()) && brokerId != item.getKey());
            //判断当前是不是第一次注册
            String oldBrokerAddr = brokerAddrsMap.get(brokerId);
            //判断当前注册的地址和之前的地址是否一致，如果不一致，比较数据版本
            if (null != oldBrokerAddr && !oldBrokerAddr.equals(brokerAddr)) {
                //注意，这个时候的情况就是当前注册过来的Broker节点的Id和brokerAddrsMap中存储的尚未更新的Broker节点信息的Id一致，但是网络地址不一致
                //这种情况下，就需要把brokerAddrsMap中存储的旧地址更新成新地址，但是这里先做了一个判断，
                //那就是判断当前注册过来的Broker节点的数据版本是否比brokerAddrsMap中存储的旧数据版本新，所谓数据其实就是主题配置信息等等
                //这里从brokerLiveTable中获取到的BrokerLiveInfo对象，其实就是当前注册过来的Broker节点对应的活跃信息，包括最后一次更新时间、心跳超时时间、数据版本、通道信息、高可用服务器地址等
                BrokerAddrInfo oldBrokerAddrInfo = new BrokerAddrInfo(clusterName, oldBrokerAddr);
                BrokerLiveInfo oldBrokerLiveInfo = this.brokerLiveTable.get(oldBrokerAddrInfo);
                if (oldBrokerLiveInfo != null) {
                    long oldStateVersion = oldBrokerLiveInfo.getDataVersion().getStateVersion();
                    //得到新注册过来的Broker节点的数据版本
                    long newStateVersion = topicConfigWrapper.getDataVersion().getStateVersion();
                    //如果新的数据版本比旧的数据版本小，那就说明新注册的Broker节点的数据版本是旧的，这种情况下就不需要更新brokerAddrsMap中存储的旧地址

                    if (oldStateVersion > newStateVersion) {
                        //记录告警日志
                        log.warn("Registered Broker conflicts with the existed one, just ignore.: Cluster:{}, BrokerName:{}, BrokerId:{}, " +
                                        "Old BrokerAddr:{}, Old Version:{}, New BrokerAddr:{}, New Version:{}.",
                                clusterName, brokerName, brokerId, oldBrokerAddr, oldStateVersion, brokerAddr, newStateVersion);
                        //从brokerLiveTable中移除当前注册过来的Broker节点的活跃信息
                        brokerLiveTable.remove(new BrokerAddrInfo(clusterName, brokerAddr));
                        return result;
                    }
                }
            }
            //如果brokerAddrsMap中没有存储当前注册过来的Broker节点的地址信息，这意味着当前注册的Broker节点是第一次注册
            //但是接下来有判断了一下Broker是否注册过来了主题信息，
            // 并且主题信息只有一个，如果是这种情况，
            // 那就记录Broker节点尚未注册的告警日志
            //我不明白这是为什么，当Broker节点启动之后，
            // 第一次注册到Namesrv节点中时，它就携带着很多系统内置的主题配置信息，所以就算是第一次注册，程序也不会执行到下面的分支
            //那什么情况下Broker节点第一次注册到Namesrv节点中时，它携带的主题配置信息只有一个呢？这个我目前也不清楚，大家有知道的可以帮我解惑
            if (!brokerAddrsMap.containsKey(brokerId)
                    && topicConfigWrapper.getTopicConfigTable().size() == 1) {
                //记录告警日志
                log.warn("Can't register topicConfigWrapper={} because broker[{}]={} has not registered.", topicConfigWrapper.getTopicConfigTable(), brokerId, brokerAddr);
                return null;
            }


            //把最新注册过来的Broker节点的Id和地址信息存储到brokerAddrsMap中
            //到此为止，brokerAddrTable这个成员变量中存储的内容已经更新完毕了
            String oldAddr = brokerAddrsMap.put(brokerId, brokerAddr);
            //接下来就是保存Broker注册过来的主题配置信息的操作了
            //先判断当前注册过来的Broker节点是否为第一次注册
            registerFirst = registerFirst || (StringUtils.isEmpty(oldAddr));
            //判断当前节点是否为主节点
            boolean isMaster = MixAll.MASTER_ID == brokerId;
            //然后再判断当前注册过来的Broker节点是否为主要从节点，所谓主要从节点其实就是Broker集群中的从节点，但是这个从节点是Broker集群中存活的从节点中Id最小的那个
            boolean isPrimeSlave = !isOldVersionBroker && !isMaster && brokerId == Collections.min(brokerAddrsMap.keySet());
            //接下来就是真正保存Broker注册过来的主题配置信息的操作了，根据下面的这个判断，只有主节点或者是主要从节点注册过来的主题配置信息才会被保存
            //因为创建主题配置信息的时候，也都是向Broker集群所在的主节点创建的
            if (null != topicConfigWrapper && (isMaster || isPrimeSlave)) {
                //注册上来的主题配置信息map
                ConcurrentMap<String, TopicConfig> topicConfigTable = topicConfigWrapper.getTopicConfigTable();
                if (topicConfigTable != null) {
                    for (Map.Entry<String, TopicConfig> entry : topicConfigTable.entrySet()) {
                        //判断是否为首次注册或者主题配置是否发生变化了，如果是第一次注册或者是主题配置信息发生了变化，那就创建或更新主题配置信息
                        if (registerFirst || this.isTopicConfigChanged(clusterName, brokerAddr, topicConfigWrapper.getDataVersion(), brokerName, entry.getValue().getTopicName())) {
                            TopicConfig topicConfig = entry.getValue();
                            //如果当前注册过来的Broker节点是主要从节点，并且Broker节点启用了ActingMaster模式，那就把主要从节点的写权限修改为只读
                            if (isPrimeSlave && brokerData.isEnableActingMaster()) {
                                //设置主要从节点的写权限为只读，从节点只能读取数据，不能写入数据
                                topicConfig.setPerm(topicConfig.getPerm() & (~PermName.PERM_WRITE));
                            }
                            //在这里为这个主题创建或者更新主题配置信息
                            this.createAndUpdateQueueData(brokerName, topicConfig);
                        }
                    }
                }
            }


            //接下来创建一个BrokerAddrInfo对象，用来存储当前注册过来的Broker节点的地址信息和所在集群的名称
            BrokerAddrInfo brokerAddrInfo = new BrokerAddrInfo(clusterName, brokerAddr);
            //根据刚刚创建好的BrokerAddrInfo对象从brokerLiveTable中获取BrokerLiveInfo对象
            //这里返回的就是当前注册过来的Broker节点的活跃信息，并且是旧的活跃信息
            //如果这里得到的是null，说明之前brokerLiveTable中没有存储过这个Broker节点的活跃信息，也就意味着现在是当前Broker节点第一次注册到Namesrv节点中
            BrokerLiveInfo prevBrokerLiveInfo = this.brokerLiveTable.put(brokerAddrInfo,
                    new BrokerLiveInfo(
                            //设置当前时间戳为这个Broker节点的最新被更新的时间
                            System.currentTimeMillis(),
                            //设置心跳超时时间，如果最新更新时间加上这个心跳超时时间小于当前时间，说明这个Broker节点已经不活跃了
                            //从这里也可以看出来，Namesrv判断Broker节点是否活跃的依据就是看这个Broker节点的最新更新时间和心跳超时时间的和是否小于当前时间
                            //所以每当Broker节点发送心跳包的时候，Namesrv节点就会更新这个Broker节点的最新更新时间
                            //但是在第五版本代码中，我还没有实现Broker节点向Namesrv发送心跳消息的功能，所以大家先记理解这个操作的作用即可，第六版本就会为大家实现Broker和Namesrv心跳检测功能了
                            timeoutMillis == null ? DEFAULT_BROKER_CHANNEL_EXPIRED_TIME : timeoutMillis,
                            //更新Broker的数据版本
                            topicConfigWrapper == null ? new DataVersion() : topicConfigWrapper.getDataVersion(),
                            //设置当前Broker节点的channel
                            channel,
                            //设置高可用服务器地址
                            //注意，每一个Broker节点注册信息到Namesrv的时候，都会把自己的高可用服务器地址注册到Namesrv中
                            //但是在Broker模块我添加了相关注释，所谓的高可用服务器地址，其实就是Broker主节点给从节点同步消息的地址
                            //如果注册信息过来的Broker节点是一个从节点，那么这个高可用服务器地址还有什么作用呢？大家先记住这个问题，在这个方法的结尾就会给大家展示答案
                            haServerAddr));
            //如果得到的prevBrokerLiveInfo为空，说明当前注册的Broker节点是第一次注册到Namesrv节点中
            if (null == prevBrokerLiveInfo) {
                //记录日志信息即可
                log.info("new broker registered, {} HAService: {}", brokerAddrInfo, haServerAddr);
            }

            //接下来要返回给Broker节点响应了，这时候就要执行一个很重要的操作
            //那就是判断当前注册信息过来的Broker节点是否为自己所在集群的主节点
            //也就是判断它的Id是不是0
            if (MixAll.MASTER_ID != brokerId) {
                //如果不是主节点，那就从自己所在集群的brokerData信息中获得主节点的网络地址
                String masterAddr = brokerData.getBrokerAddrs().get(MixAll.MASTER_ID);
                if (masterAddr != null) {
                    BrokerAddrInfo masterAddrInfo = new BrokerAddrInfo(clusterName, masterAddr);
                    //这里得到集群主节点的高可用地址，刚才我们不是把Broker节点高可用地址设置到了BrokerLiveInfo对象中了吗？现在就用上了
                    BrokerLiveInfo masterLiveInfo = this.brokerLiveTable.get(masterAddrInfo);
                    if (masterLiveInfo != null) {
                        //最后就可以把集群主节点的地址和高可用地址设置到响应中，回复给注册信息过来的Broker从节点
                        //这样一来从节点就从Namesrv中获取到了自己所在集群的主节点的地址和高可用地址
                        result.setHaServerAddr(masterLiveInfo.getHaServerAddr());
                        result.setMasterAddr(masterAddr);
                    }
                }
            }
            //如果最小的Broker节点的Id发生了变化，那就通知Broker所在集群其他节点最小的Broker节点的Id发生了变化
            if (isMinBrokerIdChanged && namesrvConfig.isNotifyMinBrokerIdChanged()) {
                notifyMinBrokerIdChanged(brokerAddrsMap, null, this.brokerLiveTable.get(brokerAddrInfo).getHaServerAddr());
            }
        } catch (Exception e) {
            log.error("registerBroker Exception", e);
        } finally {
            this.lock.writeLock().unlock();
        }//这是我自己添加的一行代码，用来验证测试类的正确性
        System.out.println("nameserv接收到了broker的注册信息" + brokerAddrTable);
        return result;
    }

    private void createAndUpdateQueueData(String brokerName, TopicConfig topicConfig) {
        //创建一个QueueData对象，这个对象就是用来封装主题队列信息的
        QueueData queueData = new QueueData();
        //设置Broker名称
        queueData.setBrokerName(brokerName);
        //设置主题写队列数量
        queueData.setWriteQueueNums(topicConfig.getWriteQueueNums());
        //设置主题读队列数量
        queueData.setReadQueueNums(topicConfig.getReadQueueNums());
        //设置主题队列的权限
        queueData.setPerm(topicConfig.getPerm());
        //设置系统标志
        queueData.setTopicSysFlag(topicConfig.getTopicSysFlag());
        //从主题配置map中获取主题配置
        Map<String, QueueData> queueDataMap = this.topicQueueTable.get(topicConfig.getTopicName());
        if (queueDataMap == null) {
            queueDataMap = new HashMap<String, QueueData>();
            queueDataMap.put(brokerName, queueData);
            this.topicQueueTable.put(topicConfig.getTopicName(), queueDataMap);
            log.info("new topic registered, {} {}", topicConfig.getTopicName(), queueData);
        } else {
            //执行到这里就意味着当前主题不是第一次注册到Namesrv节点中，那接下来就是更新主题队列信息的操作了
            //根据Broker名称从queueDataMap中获取Broker节点的队列信息
            final QueueData existedQD = queueDataMap.get(brokerName);
            if (existedQD == null) {
                //如果获取到的队列信息为空，说明当前Broker节点是第一次注册到这个主题下
                //那就把当前Broker节点的队列信息存储到queueDataMap中
                queueDataMap.put(brokerName, queueData);
            } else if (!existedQD.equals(queueData)) {
                //如果获取到的队列信息不为空，但是和当前Broker节点的队列信息不一致，那就更新队列信息
                log.info("topic changed, {} OLD: {} NEW: {}", topicConfig.getTopicName(), existedQD, queueData);
                //在这里执行更新操作
                queueDataMap.put(brokerName, queueData);
            }
        }


    }

    public boolean isTopicConfigChanged(final String clusterName, final String brokerAddr, final DataVersion dataVersion, String brokerName, String topic) {
        //判断指定的Broker集群中的主题路由信息是否发生变化
        boolean isChange = isBrokerTopicConfigChanged(clusterName, brokerAddr, dataVersion);
        //如果发生了变化，那就返回true
        if (isChange) {
            return true;
        }
        //如果没有发生变化，那就根据主题名称得到对应的主题队列数据
        final Map<String, QueueData> queueDataMap = this.topicQueueTable.get(topic);
        //对主题队列数据执行判空操作，如果为空，那就返回true
        if (queueDataMap == null || queueDataMap.isEmpty()) {
            return true;
        }
        //最后判断主题队列数据中是否包含指定的Broker名称，如果不包含，那就返回true
        //正因为不包含，而现在又要注册这个Broker节点的主题路由信息，所以返回true，表示主题路由信息发生了变化，需要更新
        return !queueDataMap.containsKey(brokerName);
    }


    public boolean isBrokerTopicConfigChanged(final String clusterName, final String brokerAddr, final DataVersion dataVersion) {
        //查询指定的Broker节点的主题配置数据版本
        DataVersion prev = queryBrokerTopicConfig(clusterName, brokerAddr);
        //判断数据版本是否发生了变化
        return null == prev || !prev.equals(dataVersion);
    }


    public DataVersion queryBrokerTopicConfig(final String clusterName, final String brokerAddr) {
        //根据Broker节点的网络地址和所在集群名称创建一个BrokerAddrInfo对象
        BrokerAddrInfo addrInfo = new BrokerAddrInfo(clusterName, brokerAddr);
        //从brokerLiveTable中获取指定地址信息的BrokerLiveInfo对象
        BrokerLiveInfo prev = this.brokerLiveTable.get(addrInfo);
        //从BrokerLiveInfo对象中获取主题配置的数据版本
        if (prev != null) {
            return prev.getDataVersion();
        }
        return null;
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @date:2025/1/10
     * @方法描述：向批量注销服务组件提交注销Broker请求的方法
     */
    public boolean submitUnRegisterBrokerRequest(UnRegisterBrokerRequestHeader unRegisterRequest) {
        return this.unRegisterService.submit(unRegisterRequest);
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @date:2025/1/10
     * @方法描述：注销Broker节点信息的方法
     */
    public void unRegisterBroker(Set<UnRegisterBrokerRequestHeader> unRegisterRequests) {
        try {
            //记录要移除的Broker名称的集合
            Set<String> removedBroker = new HashSet<>();
            //记录要减少的Broker名称的集合
            //这里我要现为大家解释一下什么是要移除的Broker节点，什么是要减少的Broker节点，请大家思考一下，如果我们构建了一个Broker集群，集群中有多个Broker节点
            //现在只是集群中的某个Broker节点信息要注销了，这个时候我们就称要执行的操作是减少某个Broker节点的信息，如果这个集群中所有节点都要注销了，那么现在我们就称要从Namesrv中移除这个Broker集群了
            //因为Broker集群节点名称都一样，所以直接用刚才创建的集合记录要移除的Broker节点名称即可。那为什么这么做呢？很简单，Broker不仅仅要注册自身的信息，还会把主题信息也注册到Namesrv中
            //假如Beoker集群的所有节点都注销了，那么这个Broker集群的主题信息也就没有了，所以这个时候就要把这个Broker集群的主题信息也从Namesrv中移除，虽然第六版本才会实现这个功能，但我相信使用过RocketMq的朋友应该都能理解这里的意思
            //如果这个Broker集群中还有节点没有注销，那么就把要注销的Broker节点名称记录到reducedBroker集合中，这样一来在根据这个集合中的名称去清理对应的主题信息时，并不会把这个集群的主题信息清理了
            //而是判断这个集群中的主节点有没有注销，并且是否开启了ActingMaster模式，如果主节点已经注销，并且集群开启了ActingMaster模式，那么就不会清理这个集群的主题信息，而是把主题队列的写权限取消了
            //大家可以先记住这个逻辑，等到第六版本的代码中我会为大家详细讲解这个逻辑
            Set<String> reducedBroker = new HashSet<>();
            //记录需要通知的Broker节点的信息，这个map的作用在第五版本也体现不出来，等到引入Broker集群了才能体现出来，我这里之所以没有把这个map和后面相关的代码都删除
            //是想给大家稍微延伸一点，我知道有的朋友已经研究过MQ的源码了，也许还想了解得更深，所以我也愿意在适当的时候为大家提供一些思考的空间
            //请大家想一想，假如一个Broker集群的主节点注销了，而这个主节点所在的集群又启动了ActingMaster模式，这时候就要执行一个最重要的操作
            //那就是选取一个从节点，让从节点代理主节点的工作，这个时候就需要通知这个集群中的其他节点，让它们知道主节点已经注销了，从节点要代理主节点的工作
            Map<String, BrokerStatusChangeInfo> needNotifyBrokerMap = new HashMap<>();
            this.lock.writeLock().lockInterruptibly();
            //遍历要注销Broker信息的请求
            for (final UnRegisterBrokerRequestHeader unRegisterRequest : unRegisterRequests) {
                //得到要注销的Broker节点的名称
                final String brokerName = unRegisterRequest.getBrokerName();
                //得到要注销的Broker节点所在集群的名称
                final String clusterName = unRegisterRequest.getClusterName();
                //得到要注销的Broker节点的网络地址
                final String brokerAddr = unRegisterRequest.getBrokerAddr();
                //创建一个BrokerAddrInfo对象，用来存储要注销的Broker节点的网络地址和所在集群的名称
                BrokerAddrInfo brokerAddrInfo = new BrokerAddrInfo(clusterName, brokerAddr);
                //因为这个Broker要注销了，不必再判断是否心跳过期了，所以要从brokerLiveTable中移除这个Broker节点的活跃信息
                BrokerLiveInfo brokerLiveInfo = this.brokerLiveTable.remove(brokerAddrInfo);
                log.info("unregisterBroker, remove from brokerLiveTable {}, {}", brokerLiveInfo != null ? "OK" : "Failed", brokerAddrInfo);

                //定义一个变量，表示是否要移除Broker的名称，如果Broker所在集群的所有Broker节点都注销了，这个时候就要移除了
                boolean removeBrokerName = false;
                //定义一个变量，表示集群中最小的BrokerId是否发生了变化，如果最小的BrokerId发生了变化，那很可能和主节点的切换有关
                boolean isMinBrokerIdChanged = false;
                //根据当前要注销的Broker节点的名称从brokerAddrTable中获取BrokerData对象，这时候就得到了当前Broker所在集群的节点的所有信息
                BrokerData brokerData = this.brokerAddrTable.get(brokerName);
                //对BrokerData对象进行判空处理
                if (null != brokerData) {
                    //判断要注销的Broker节点Id是不是当前集群中最小的Broker节点Id，如果是，那就更新isMinBrokerIdChanged为true
                    if (!brokerData.getBrokerAddrs().isEmpty() && unRegisterRequest.getBrokerId().equals(Collections.min(brokerData.getBrokerAddrs().keySet()))) {
                        //更新isMinBrokerIdChanged为true
                        isMinBrokerIdChanged = true;
                    }
                    //从BrokerData对象移除brokerAddr，这里的意思就是把当前要注销的Broker节点的地址信息从BrokerData对象的brokerAddrs成员变量中移除
                    boolean removed = brokerData.getBrokerAddrs().entrySet().removeIf(item -> item.getValue().equals(brokerAddr));
                    log.info("unregisterBroker, remove addr from brokerAddrTable {}, {}", removed ? "OK" : "Failed", brokerAddrInfo);
                    //移除了对应的Broker节点的地址信息之后，判断brokerData中是否还有Broker节点的地址信息
                    if (brokerData.getBrokerAddrs().isEmpty()) {
                        //如果没有则意味着Broker所在集群的所有节点都注销了，这个时候就可以把这个Broker集群的信息从brokerAddrTable中移除了
                        this.brokerAddrTable.remove(brokerName);
                        log.info("unregisterBroker, remove name from brokerAddrTable OK, {}", brokerName);
                        //更新removeBrokerName为true
                        removeBrokerName = true;
                    } else if (isMinBrokerIdChanged) {
                        //如果Broker集群中还有其它节点没有注销，并且最小的BrokerId发生了变化，那就需要通知这个集群中的其他节点，让它们知道主节点已经切换了
                        //这里创建了一个BrokerStatusChangeInfo对象，这个对象内部封装者当前处理的Broker节点所在集群中其他节点的地址信息，以及当前处理的Broker节点的地址信息
                        //因为注销当前节点触发了集群最小BrokerId变更，那么当前注销的节点就是下线的主节点
                        needNotifyBrokerMap.put(brokerName, new BrokerStatusChangeInfo(brokerData.getBrokerAddrs(), brokerAddr, null));
                    }
                }
                if (removeBrokerName) {
                    //判断是否有Broker集群的信息需要从Namesrv中移除
                    //如果有则先得到这个Broker集群的所有节点的名称
                    Set<String> nameSet = this.clusterAddrTable.get(clusterName);
                    if (nameSet != null) {
                        //从集合中移除当前要注销的Broker节点的名称
                        boolean removed = nameSet.remove(brokerName);
                        log.info("unregisterBroker, remove name from clusterAddrTable {}, {}", removed ? "OK" : "Failed", brokerName);
                        if (nameSet.isEmpty()) {
                            //如果集合为空了，意味着这个Broker集群的所有节点都注销了，这个时候就可以把这个Broker集群的信息从clusterAddrTable中移除了
                            this.clusterAddrTable.remove(clusterName);
                            log.info("unregisterBroker, remove cluster from clusterAddrTable {}", clusterName);
                        }
                    }
                    //将当前Broker名称添加到移除的Broker集合中
                    removedBroker.add(brokerName);
                } else {
                    //执行到这里意味着当前Broker集群中还有节点没有注销，那就把当前要注销的Broker节点的名称添加到减少的Broker集合中
                    reducedBroker.add(brokerName);
                }
            }

            //接下来就可以执行清理对应的主题信息的操作了，但这个操作在第六版本才会实现，所以现在先把代码注释了
            //清理的时候就会用到一开始定义的两个集合了，具体使用方法到第六版本代码大家就清楚了
            //cleanTopicByUnRegisterRequests(removedBroker, reducedBroker);
            //如果需要通知的Broker状态变化信息不为空，那就通知当前住校的Broker所在集群中最小BrokerId发生了变化
            if (!needNotifyBrokerMap.isEmpty() && namesrvConfig.isNotifyMinBrokerIdChanged()) {
                //通知Broker集群中最小BrokerId发生了变化
                notifyMinBrokerIdChanged(needNotifyBrokerMap);
            }
        } catch (Exception e) {
            log.error("unregisterBroker Exception", e);
        } finally {
            this.lock.writeLock().unlock();
        }
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @date:2025/1/10
     * @方法描述：通知指定的Broker集群其他节点中最小BrokerID发生变化的方法。
     */
    private void notifyMinBrokerIdChanged(Map<String, BrokerStatusChangeInfo> needNotifyBrokerMap) throws InterruptedException, RemotingConnectException, RemotingTimeoutException, RemotingSendRequestException, RemotingTooMuchRequestException {

        for (String brokerName : needNotifyBrokerMap.keySet()) {
            BrokerStatusChangeInfo brokerStatusChangeInfo = needNotifyBrokerMap.get(brokerName);
            //得到要通知的Broker节点所在集群的信息
            BrokerData brokerData = brokerAddrTable.get(brokerName);
            //判断当前处理的Broker节点是否启用了ActingMaster模式
            if (brokerData != null && brokerData.isEnableActingMaster()) {
                //如果启用了那就可以通知这个集群中的其他节点，让它们知道主节点已经切换了
                notifyMinBrokerIdChanged(brokerStatusChangeInfo.getBrokerAddrs(), brokerStatusChangeInfo.getOfflineBrokerAddr(), brokerStatusChangeInfo.getHaBrokerAddr());
            }
        }
    }

    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @date:2025/1/10
     * @方法描述：通知指定的Broker集群其他节点中最小BrokerID发生变化的方法。
     */
    private void notifyMinBrokerIdChanged(Map<Long, String> brokerAddrMap, String offlineBrokerAddr, String haBrokerAddr) throws InterruptedException, RemotingSendRequestException, RemotingTimeoutException, RemotingTooMuchRequestException, RemotingConnectException {

        if (brokerAddrMap == null || brokerAddrMap.isEmpty() || this.namesrvController == null) {
            return;
        }
        //创建通知最小BrokerId发生变化的请求头对象
        NotifyMinBrokerIdChangeRequestHeader requestHeader = new NotifyMinBrokerIdChangeRequestHeader();
        //得到Broker所在集群目前最小的BrokerId
        long minBrokerId = Collections.min(brokerAddrMap.keySet());
        //设置请求头中的最小BrokerId
        requestHeader.setMinBrokerId(minBrokerId);
        //设置请求头中最小BrokerId的地址
        requestHeader.setMinBrokerAddr(brokerAddrMap.get(minBrokerId));
        //设置请求头中的下线的Broker的地址
        requestHeader.setOfflineBrokerAddr(offlineBrokerAddr);
        //设置请求头中的高可用Broker的地址，这里其实是null
        requestHeader.setHaBrokerAddr(haBrokerAddr);
        //选择集群中要通知的Broker地址列表
        List<String> brokerAddrsNotify = chooseBrokerAddrsToNotify(brokerAddrMap, offlineBrokerAddr);
        log.info("min broker id changed to {}, notify {}, offline broker addr {}", minBrokerId, brokerAddrsNotify, offlineBrokerAddr);
        //创建请求对象
        RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.NOTIFY_MIN_BROKER_ID_CHANGE, requestHeader);
        //遍历地址列表
        for (String brokerAddr : brokerAddrsNotify) {
            //通知每一个Broker节点
            this.namesrvController.getRemotingClient().invokeOneway(brokerAddr, request, 300);
        }
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @date:2025/1/10
     * @方法描述：选择集群中要通知的Broker地址列表的方法。
     */
    private List<String> chooseBrokerAddrsToNotify(Map<Long, String> brokerAddrMap, String offlineBrokerAddr) {
        //如果集群中剩余的Broker节点只有一个，那就通知所有的Broker节点
        if (offlineBrokerAddr != null || brokerAddrMap.size() == 1) {
            return new ArrayList<>(brokerAddrMap.values());
        }
        //得到集群中最小的BrokerId
        long minBrokerId = Collections.min(brokerAddrMap.keySet());
        List<String> brokerAddrList = new ArrayList<>();
        for (Long brokerId : brokerAddrMap.keySet()) {
            //通知除了最小的BrokerId之外的所有Broker节点
            if (brokerId != minBrokerId) {
                brokerAddrList.add(brokerAddrMap.get(brokerId));
            }
        }
        return brokerAddrList;
    }

    public TopicRouteData pickupTopicRouteData(final String topic) {
        //创建一个TopicRouteData对象用来存储主题路由信息
        TopicRouteData topicRouteData = new TopicRouteData();
        //定义一个是否找到了队列信息的标记
        boolean foundQueueData = false;
        //定义一个是否找到了Broker集群信息的标记
        boolean foundBrokerData = false;
        //创建一个BrokerData集合，用来存储QueueData所属的BrokerData
        List<BrokerData> brokerDataList = new LinkedList<>();
        //这里就已经把BrokerData集合设置到了TopicRouteData对象中了
        topicRouteData.setBrokerDatas(brokerDataList);

        try {
            this.lock.readLock().lockInterruptibly();
            Map<String, QueueData> queueDataMap = this.topicQueueTable.get(topic);
            log.info("get route info for the topic: " + topic);
            if (queueDataMap != null) {
                topicRouteData.setQueueDatas(new ArrayList<>(queueDataMap.values()));
                //更新找到队列的额标记
                foundQueueData = true;
                //queueDataMap.keySet()得到的就是该主题下可以路由的所有队列所属的BrokerName集合
                Set<String> brokerNameSet = new HashSet<>(queueDataMap.keySet());
                for (String brokerName : brokerNameSet) {
                    //根据BrokerName从brokerAddrTable中获取BrokerData对象
                    BrokerData brokerData = this.brokerAddrTable.get(brokerName);
                    if (null == brokerData) {
                        continue;
                    }
                    //克隆BrokerData对象
                    BrokerData brokerDataClone = new BrokerData(brokerData);
                    //把克隆后的BrokerData对象添加到之前创建的brokerDataList集合中
                    brokerDataList.add(brokerDataClone);
                    //更新找到Broker集群信息的标记
                    foundBrokerData = true;

                }
            }
        } catch (Exception e) {
            log.error("pickupTopicRouteData Exception", e);
        } finally {
            this.lock.readLock().unlock();

        }
        log.debug("pickupTopicRouteData brokerDataList: {}", brokerDataList);
        //以上操作结束之后，就得到了一个完整的TopicRouteData对象了，并且这个对象中包含了主题的路由信息
        //路由信息分为两部分，一部分是Broker集群信息，一部分是QueueData队列信息
        //QueueData队列信息可以知道这个队列所属的Broker集群是哪个，而通过Broker集群信息，也就是BrokerData对象
        //可以知道这个Broker集群下有哪些Broker节点，接下来就可以根据这些信息来进行路由了
        //当然，在第七版本代码中大家还暂时不用深入思考这里的逻辑，也不用深入思考TopicRouteData信息返回给客户端之后应该怎么被使用
        //因为我们是基于5.2.0版本的RocketMq源码进行讲解的，新版本的框架引入了Proxy模块，等实现了后面代码之后
        //大家就会发现，不管Namesrv返回了什么路由信息，最后生产者消费者得到的Broker地址实际上都是Proxy服务器的地址，这里面的细节大家很快就会明白的
        //当然，有一点大家现在肯定意识到了，不管怎么路由，最后肯定是把Broker所在集群主节点的地址返回给客户端的，这个是肯定的
        //接下来就是返回这个TopicRouteData对象的前置操作了
        if (foundBrokerData && foundQueueData) {//topicRouteData.setTopicQueueMappingByBroker(this.topicQueueMappingInfoTable.get(topic));
            //如果Namesrv不支持Acting Master模式，那就直接返回主题路由信息
            if (!namesrvConfig.isSupportActingMaster()) {
                return topicRouteData;
            }
            //如果主题名称是以系统主题前缀开头的，直接返回主题路由数据
            if (topic.startsWith(TopicValidator.SYNC_BROKER_MEMBER_GROUP_PREFIX)) {
                return topicRouteData;
            }

            //如果要返回的主题路由数据中没有Broker集群信息或者队列信息，直接返回这个主题路由数据即可
            if (topicRouteData.getBrokerDatas().size() == 0 || topicRouteData.getQueueDatas().size() == 0) {
                return topicRouteData;
            }

            //定义一个是否需要使用ActingMaster模式的标记
            boolean needActingMaster = false;
            //遍历主题路由信息中的BrokerData对象
            for (final BrokerData brokerData : topicRouteData.getBrokerDatas()) {
                //判断每一个Broker集群中是否有主节点地址，如果没有，那就需要使用ActingMaster模式
                //看到这里也许大家的思维已经混乱了，不知道这是什么意思，接下来就让我为大家简单梳理一下，一直以来我都跟大家说，我们目前构架框架都是在单集群单主节点模式下运行的
                //现在我们改变一下这个规则，假如现在可以构建集群，并且集群中允许存在多个主节点，也就是说一个Broker集群中可以有多个主节点，也就意味着一个Broker集群可以存在多个master-slave组
                //现在我们得到了一个TopicRouteData对象，这个对象中包含了两部分信息，一部分是Broker组信息，一部分是QueueData队列信息
                //QueueData队列和Broker组是一一对应的，也就是说，一个QueueData队列属于一个Broker组
                //并且QueueData队列中有一个字段是BrokerName，这个字段就是这个QueueData队列所属的Broker组的名称
                //根据QueueData队列中的BrokerName字段，我们可以从Broker组信息中找到对应的BrokerData对象，而找到了BrokerData对象之后，就可以从该对象中找到Broker组中主节点的地址
                //这个Broker主节点的地址就是我们最终要返回给客户端的Broker地址
                //比如说现在有一个Test主题，这个主题在BrokerA和BrokerB上都创建了
                //那么Namesrv的topicQueueTable中就会存储以下数据
                //
                //key               value
                //String            Map
                //
                //                  BrokerA -> QueueData
                //Test
                //                  BrokerB -> QueueData
                if (brokerData.getBrokerAddrs().size() != 0 && !brokerData.getBrokerAddrs().containsKey(MixAll.MASTER_ID)) {
                    //更新需要使用ActingMaster模式的标记
                    needActingMaster = true;
                    break;
                }
            }
            //如果不需要使用ActingMaster模式，直接返回主题路由信息
            if (!needActingMaster) {
                return topicRouteData;
            }

            //执行到这里意味着需要使用ActingMaster模式
            for (final BrokerData brokerData : topicRouteData.getBrokerDatas()) {
                //那就开始遍历主题路由信息中的BrokerData对象，也就是遍历每一个Broker组
                final HashMap<Long, String> brokerAddrs = brokerData.getBrokerAddrs();
                //如果当前遍历到的Broker组中有主节点，那就不需要设置Acting Master模式，如果没有主节点，但是不支持Acting Master模式，那也不需要设置Acting Master模式
                if (brokerAddrs.size() == 0 || brokerAddrs.containsKey(MixAll.MASTER_ID) || !brokerData.isEnableActingMaster()) {
                    //直接跳过本次循环即可
                    continue;
                }
                //执行到这里意味着需要使用ActingMaster模式
                for (final QueueData queueData : topicRouteData.getQueueDatas()) {
                    //主题路由队列所对应的Broker组的名称和当前遍历到的Broker组的名称相等
                    if (queueData.getBrokerName().equals(brokerData.getBrokerName())) {
                        //接下来就可以设置了，因为当前Broker组中没有主节点，所以需要设置Acting Master模式
                        //判断一下当前队列的权限，如果是不可写的，那就不需要设置Acting Master模式
                        if (!PermName.isWriteable(queueData.getPerm())) {
                            //如果是可写的，就得到当前Broker组中最小Id
                            final Long minBrokerId = Collections.min(brokerAddrs.keySet());
                            //先把最小Id对应的地址移除
                            final String actingMasterAddr = brokerAddrs.remove(minBrokerId);
                            //然后把最小Id对应的地址设置为主节点地址
                            //让这个最小Id的从节点代替主节点工作，但是也仅仅限于消费者消费消息，因为现在队列权限是不可写的，所以生产者无法发送消息
                            brokerAddrs.put(MixAll.MASTER_ID, actingMasterAddr);
                        }
                        break;
                    }
                }
            }
            return topicRouteData;
        }
        return null;
    }

    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @date:2025/1/10
     * @方法描述：定期扫描心跳超时的Broker节点的方法。
     */
    public void scanNotActiveBroker() {
        try {
            log.info("start scanNotActiveBroker");
            for (Map.Entry<BrokerAddrInfo, BrokerLiveInfo> next : this.brokerLiveTable.entrySet()) {
                //得到Broker最后收到心跳的时间
                long last = next.getValue().getLastUpdateTimestamp();
                //得到当前Broker的心跳超时时间
                long timeoutMillis = next.getValue().getHeartbeatTimeoutMillis();
                //如果最后一次心跳时间加上心跳超时时间小于当前时间，那就关闭这个Broker的通信通道
                if ((last + timeoutMillis) < System.currentTimeMillis()) {
                    //channel心跳超时了，直接关闭channel
                    RemotingHelper.closeChannel(next.getValue().getChannel());
                    log.warn("The broker channel expired, {} {}ms", next.getKey(), timeoutMillis);
                    //执行channel销毁操作，这里传递进去的就是要销毁的Broker节点的地址信息
                    this.onChannelDestroy(next.getKey());
                }
            }
        } catch (Exception e) {
            log.error("scanNotActiveBroker exception", e);
        }
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @date:2025/1/10
     * @方法描述：销毁与Broker节点对应的channel的方法
     */
    public void onChannelDestroy(BrokerAddrInfo brokerAddrInfo) {
        //创建要注销Broker的请求头
        UnRegisterBrokerRequestHeader unRegisterRequest = new UnRegisterBrokerRequestHeader();
        //定义一个是否要注销Broker的标记
        boolean needUnRegister = false;
        if (brokerAddrInfo != null) {
            try {
                try {
                    this.lock.readLock().lockInterruptibly();
                    //把相关信息填充到请求头中，反回是否需要注销的结果
                    needUnRegister = setupUnRegisterRequest(unRegisterRequest, brokerAddrInfo);
                } finally {
                    this.lock.readLock().unlock();
                }
            } catch (Exception e) {
                log.error("onChannelDestroy Exception", e);
            }
        }
        //如果需要注销
        if (needUnRegister) {
            //就把注销请求提交给批量注销服务组件
            boolean result = this.submitUnRegisterBrokerRequest(unRegisterRequest);
            log.info("the broker's channel destroyed, submit the unregister request at once, " +
                    "broker info: {}, submit result: {}", unRegisterRequest, result);
        }
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @date:2025/1/10
     * @方法描述：销毁指定Broker节点的channel的方法，该方法会在BrokerHousekeepingService对象中被调用
     */
    public void onChannelDestroy(Channel channel) {
        UnRegisterBrokerRequestHeader unRegisterRequest = new UnRegisterBrokerRequestHeader();
        BrokerAddrInfo brokerAddrFound = null;
        boolean needUnRegister = false;
        if (channel != null) {
            try {
                try {
                    this.lock.readLock().lockInterruptibly();
                    for (Map.Entry<BrokerAddrInfo, BrokerLiveInfo> entry : this.brokerLiveTable.entrySet()) {
                        //如果当前要销毁的channel和BrokerLiveInfo中遍历到的channel相等，那就找到了要销毁的Broker节点的地址信息
                        if (entry.getValue().getChannel() == channel) {
                            //得到要销毁的Broker节点的地址信息
                            brokerAddrFound = entry.getKey();
                            break;
                        }
                    }
                    if (brokerAddrFound != null) {
                        //填充注销请求头
                        needUnRegister = setupUnRegisterRequest(unRegisterRequest, brokerAddrFound);
                    }
                } finally {
                    this.lock.readLock().unlock();
                }
            } catch (Exception e) {
                log.error("onChannelDestroy Exception", e);
            }
        }
        if (needUnRegister) {
            //如果要注销则把请求头提交给批量注销服务组件
            boolean result = this.submitUnRegisterBrokerRequest(unRegisterRequest);
            log.info("the broker's channel destroyed, submit the unregister request at once, " +
                    "broker info: {}, submit result: {}", unRegisterRequest, result);
        }
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @date:2025/1/10
     * @方法描述：设置注销请求头的方法
     */
    private boolean setupUnRegisterRequest(UnRegisterBrokerRequestHeader unRegisterRequest, BrokerAddrInfo brokerAddrInfo) {
        //设置注销请求的Broker节点所在集群的名称
        unRegisterRequest.setClusterName(brokerAddrInfo.getClusterName());
        //设置要注销的Broker节点的地址
        unRegisterRequest.setBrokerAddr(brokerAddrInfo.getBrokerAddr());
        for (Map.Entry<String, BrokerData> stringBrokerDataEntry : this.brokerAddrTable.entrySet()) {
            BrokerData brokerData = stringBrokerDataEntry.getValue();
            if (!brokerAddrInfo.getClusterName().equals(brokerData.getCluster())) {
                continue;
            }
            //如果BrokerData对象中的Broker所在集群的名称和要注销的Broker所在集群名称相等，那就设置注销请求的Broker名称和BrokerId
            for (Map.Entry<Long, String> entry : brokerData.getBrokerAddrs().entrySet()) {
                Long brokerId = entry.getKey();
                String brokerAddr = entry.getValue();
                //根据网络地址找到了要注销的Broker节点
                if (brokerAddr.equals(brokerAddrInfo.getBrokerAddr())) {
                    //设置注销请求的Broker名称
                    unRegisterRequest.setBrokerName(brokerData.getBrokerName());
                    //设置注销请求的BrokerId
                    unRegisterRequest.setBrokerId(brokerId);
                    return true;
                }
            }
        }
        return false;
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @date:2025/1/14
     * @方法描述：得到注册到Namesrv中所有Broker集群的信息
     */
    public ClusterInfo getAllClusterInfo() {
        //创建一个ClusterInfo对象，这个对象就是用来封装所有Broker集群的信息的
        ClusterInfo clusterInfoSerializeWrapper = new ClusterInfo();
        //把所有Broker集群信息设置到ClusterInfo对象中
        clusterInfoSerializeWrapper.setBrokerAddrTable(this.brokerAddrTable);
        clusterInfoSerializeWrapper.setClusterAddrTable(this.clusterAddrTable);
        return clusterInfoSerializeWrapper;
    }

    public void registerTopic(String topic, List<QueueData> queueDatas) {
        if (queueDatas == null || queueDatas.isEmpty()) {
            return;
        }
        try {
            this.lock.writeLock().lockInterruptibly();
            if (this.topicQueueTable.containsKey(topic)) {
                Map<String, QueueData> queueDataMap = this.topicQueueTable.get(topic);
                for (QueueData queueData : queueDatas) {
                    if (!this.brokerAddrTable.containsKey(queueData.getBrokerName())) {
                        log.warn("Register topic contains illegal broker, {}, {}", topic, queueData);
                        return;
                    }
                    queueDataMap.put(queueData.getBrokerName(), queueData);
                }
                log.info("Topic route already exist.{}, {}", topic, this.topicQueueTable.get(topic));
            } else {
                Map<String, QueueData> queueDataMap = new HashMap<>();
                for (QueueData queueData : queueDatas) {
                    if (!this.brokerAddrTable.containsKey(queueData.getBrokerName())) {
                        log.warn("Register topic contains illegal broker, {}, {}", topic, queueData);
                        return;
                    }
                    queueDataMap.put(queueData.getBrokerName(), queueData);
                }
                this.topicQueueTable.put(topic, queueDataMap);
                log.info("Register topic route:{}, {}", topic, queueDatas);
            }
        } catch (Exception e) {
            log.error("registerTopic Exception", e);
        } finally {
            this.lock.writeLock().unlock();
        }
    }

    public void updateBrokerInfoUpdateTimestamp(final String clusterName, final String brokerAddr) {
        BrokerAddrInfo addrInfo = new BrokerAddrInfo(clusterName, brokerAddr);
        BrokerLiveInfo prev = this.brokerLiveTable.get(addrInfo);
        if (prev != null) {
            //在这里把BrokerLiveInfo对象中的最后更新时间戳设置为当前时间戳
            //这样一来，Namesrv心跳检测功能就可以正常工作了
            prev.setLastUpdateTimestamp(System.currentTimeMillis());
        }
    }

    public void deleteTopic(final String topic, final String clusterName) {
        try {
            this.lock.writeLock().lockInterruptibly();
            Set<String> brokerNames = this.clusterAddrTable.get(clusterName);
            if (brokerNames == null || brokerNames.isEmpty()) {
                return;
            }
            Map<String, QueueData> queueDataMap = this.topicQueueTable.get(topic);
            if (queueDataMap != null) {
                for (String brokerName : brokerNames) {
                    final QueueData removedQD = queueDataMap.remove(brokerName);
                    if (removedQD != null) {
                        log.info("deleteTopic, remove one broker's topic {} {} {}", brokerName, topic, removedQD);
                    }
                }
                if (queueDataMap.isEmpty()) {
                    log.info("deleteTopic, remove the topic all queue {} {}", clusterName, topic);
                    this.topicQueueTable.remove(topic);
                }
            }
        } catch (Exception e) {
            log.error("deleteTopic Exception", e);
        } finally {
            this.lock.writeLock().unlock();
        }
    }

    public void deleteTopic(final String topic) {
        try {
            this.lock.writeLock().lockInterruptibly();
            this.topicQueueTable.remove(topic);
        } catch (Exception e) {
            log.error("deleteTopic Exception", e);
        } finally {
            this.lock.writeLock().unlock();
        }
    }
}


//以下是RouteInfoManager类中用到的一些内部类，类的内容都非常简单，在上面代码的注释中我也详细介绍了，所以下面的代码就不添加注释了，大家自己看看就行

class BrokerAddrInfo {
    private String clusterName;
    private String brokerAddr;
    private int hash;


    public BrokerAddrInfo(String clusterName, String brokerAddr) {
        this.clusterName = clusterName;
        this.brokerAddr = brokerAddr;
    }


    public String getClusterName() {
        return clusterName;
    }


    public String getBrokerAddr() {
        return brokerAddr;
    }


    public boolean isEmpty() {
        return clusterName.isEmpty() && brokerAddr.isEmpty();
    }


    @Override
    public boolean equals(Object obj) {
        if (this == obj) {
            return true;
        }
        if (obj == null) {
            return false;
        }
        if (obj instanceof BrokerAddrInfo) {
            BrokerAddrInfo addr = (BrokerAddrInfo) obj;
            return clusterName.equals(addr.clusterName) && brokerAddr.equals(addr.brokerAddr);
        }
        return false;
    }


    @Override
    public int hashCode() {
        int h = hash;
        if (h == 0 && clusterName.length() + brokerAddr.length() > 0) {
            for (int i = 0; i < clusterName.length(); i++) {
                h = 31 * h + clusterName.charAt(i);
            }
            h = 31 * h + '_';
            for (int i = 0; i < brokerAddr.length(); i++) {
                h = 31 * h + brokerAddr.charAt(i);
            }
            hash = h;
        }
        return h;
    }


    @Override
    public String toString() {
        return "BrokerIdentityInfo [clusterName=" + clusterName + ", brokerAddr=" + brokerAddr + "]";
    }
}


class BrokerLiveInfo {
    //最新接收到心跳的时间戳
    private long lastUpdateTimestamp;
    //心跳超时时间
    private long heartbeatTimeoutMillis;
    //注册的数据的版本号
    private DataVersion dataVersion;
    //当前Broker节点的channel
    private Channel channel;
    //高可用地址
    private String haServerAddr;


    public BrokerLiveInfo(long lastUpdateTimestamp, long heartbeatTimeoutMillis, DataVersion dataVersion,
                          Channel channel, String haServerAddr) {

        this.lastUpdateTimestamp = lastUpdateTimestamp;
        this.heartbeatTimeoutMillis = heartbeatTimeoutMillis;
        this.dataVersion = dataVersion;
        this.channel = channel;
        this.haServerAddr = haServerAddr;
    }


    public long getLastUpdateTimestamp() {
        return lastUpdateTimestamp;
    }


    public void setLastUpdateTimestamp(long lastUpdateTimestamp) {
        this.lastUpdateTimestamp = lastUpdateTimestamp;
    }


    public long getHeartbeatTimeoutMillis() {
        return heartbeatTimeoutMillis;
    }


    public void setHeartbeatTimeoutMillis(long heartbeatTimeoutMillis) {
        this.heartbeatTimeoutMillis = heartbeatTimeoutMillis;
    }


    public DataVersion getDataVersion() {
        return dataVersion;
    }


    public void setDataVersion(DataVersion dataVersion) {
        this.dataVersion = dataVersion;
    }


    public Channel getChannel() {
        return channel;
    }


    public void setChannel(Channel channel) {
        this.channel = channel;
    }


    public String getHaServerAddr() {
        return haServerAddr;
    }


    public void setHaServerAddr(String haServerAddr) {
        this.haServerAddr = haServerAddr;
    }


    @Override
    public String toString() {
        return "BrokerLiveInfo [lastUpdateTimestamp=" + lastUpdateTimestamp + ", dataVersion=" + dataVersion
                + ", channel=" + channel + ", haServerAddr=" + haServerAddr + "]";
    }
}


class BrokerStatusChangeInfo {
    Map<Long, String> brokerAddrs;
    String offlineBrokerAddr;
    String haBrokerAddr;


    public BrokerStatusChangeInfo(Map<Long, String> brokerAddrs, String offlineBrokerAddr, String haBrokerAddr) {
        this.brokerAddrs = brokerAddrs;
        this.offlineBrokerAddr = offlineBrokerAddr;
        this.haBrokerAddr = haBrokerAddr;
    }


    public Map<Long, String> getBrokerAddrs() {
        return brokerAddrs;
    }


    public void setBrokerAddrs(Map<Long, String> brokerAddrs) {
        this.brokerAddrs = brokerAddrs;
    }


    public String getOfflineBrokerAddr() {
        return offlineBrokerAddr;
    }


    public void setOfflineBrokerAddr(String offlineBrokerAddr) {
        this.offlineBrokerAddr = offlineBrokerAddr;
    }


    public String getHaBrokerAddr() {
        return haBrokerAddr;
    }


    public void setHaBrokerAddr(String haBrokerAddr) {
        this.haBrokerAddr = haBrokerAddr;
    }
}
